package cn.com.visionsky.bocsz.bda.webservice.handler.history_trans_query;


import cn.com.visionsky.bocsz.bda.webservice.Constants;
import cn.com.visionsky.bocsz.bda.webservice.cache.FieldPermissionsCache;
import cn.com.visionsky.bocsz.bda.webservice.exception.RequestXmlException;
import cn.com.visionsky.bocsz.bda.webservice.handler.Handler;
import cn.com.visionsky.bocsz.bda.webservice.util.CommonUtils;
import cn.com.visionsky.bocsz.bda.webservice.ws.FullTextUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.stereotype.Component;

import javax.servlet.http.HttpServletRequest;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by 46043 on 2016/11/2.
 */
@Component
public class HistoryTransQueryHandler implements Handler {

    private static final Logger logger = LoggerFactory.getLogger(HistoryTransQueryHandler.class);
    public static final String separator = ",";

    public static final String TABLE_LSCX_BANCS_CARD_ACT_KEY_CARD = "LSCX_BANCS_CARD_ACT_KEY_CARD";
    public static final String TABLE_LSCX_OLD_NEW_ACT_KEY_NEW_ACT = "LSCX_OLD_NEW_ACT_KEY_NEW_ACT";
    public static final String TABLE_LSCX_MAIN_SUB_ACT_KEY_ACT = "LSCX_MAIN_SUB_ACT_KEY_ACT";
    public static final String TABLE_LSCX_OLD_NEW_ACT_KEY_OLD_ACT = "LSCX_OLD_NEW_ACT_KEY_OLD_ACT";
    public static final String TABLE_LSCX_SAFE_TO_RBS_KEY_OLD_ACT = "LSCX_SAFE_TO_RBS_KEY_OLD_ACT";
    public static final String TABLE_LSCX_SAFE_TO_RBS_KEY_OLD_OLD_ACT = "LSCX_SAFE_TO_RBS_KEY_OLD_OLD_ACT";
    public static final String TABLE_LSCX_TRAN = "LSCX_TRAN";
    public static final String COLUMN_FAMILY_F = "F";
    public static final byte[] COLUMN_FAMILY_F_BYTES = COLUMN_FAMILY_F.getBytes();
    public static final byte[] COLUMN_ATM_ACCOUNT_BYTES = "ATM_ACCOUNT".getBytes();
    public static final byte[] COLUMN_SUB_ACT_NO_BYTES = "SUB_ACT_NO".getBytes();
    public static final byte[] COLUMN_ACT_TYPE_BYTES = "ACT_TYPE".getBytes();
    public static final byte[] COLUMN_OLD_ACT_NO_BYTES = "OLD_ACT_NO".getBytes();
    public static final byte[] COLUMN_NEW_ACT_NO_BYTES = "NEW_ACT_NO".getBytes();
    public static final byte[] COLUMN_NEW_SUB_ACT_NO_BYTES = "NEW_SUB_ACT_NO".getBytes();
    public static final byte[] COLUMN_AC_NO_BYTES = "AC_NO".getBytes();
    public static final byte[] COLUMN_SUB_AC_NO_BYTES = "SUB_AC_NO".getBytes();
    public static final byte[] COLUMN_RBS_ACT_BYTES = "RBS_ACT".getBytes();
    public static final byte[] COLUMN_SAFE_ACT_BYTES = "SAFE_ACT".getBytes();
    public static final byte[] COLUMN_ACT_NO_BYTES = "ACT_NO".getBytes();
    public static final byte[] COLUMN_REC_NO_BYTES = "REC_NO".getBytes();
    public static final byte[] COLUMN_APPG_DATE_BYTES = "APPG_DATE".getBytes();
    public static final byte[] COLUMN_POST_DATE_BYTES = "POST_DATE".getBytes();
    public static final byte[] COLUMN_SYS_DATE_BYTES = "SYS_DATE".getBytes();
    public static final byte[] COLUMN_SYS_TIME_BYTES = "SYS_TIME".getBytes();
    public static final byte[] COLUMN_TRAN_BR_BYTES = "TRAN_BR".getBytes();
    public static final byte[] COLUMN_TRAN_TELLER_BYTES = "TRAN_TELLER".getBytes();
    public static final byte[] COLUMN_CURR_CODE_BYTES = "CURR_CODE".getBytes();
    public static final byte[] COLUMN_CURR_NAME_BYTES = "CURR_NAME".getBytes();
    public static final byte[] COLUMN_TRAN_NAME_BYTES = "TRAN_NAME".getBytes();
    public static final byte[] COLUMN_TRAN_AMT_BYTES = "TRAN_AMT".getBytes();
    public static final byte[] COLUMN_TRAN_BAL_BYTES = "TRAN_BAL".getBytes();
    public static final byte[] COLUMN_PROD_CQ_BYTES = "PROD_CQ".getBytes();
    public static final byte[] COLUMN_TRAN_MEMO_BYTES = "TRAN_MEMO".getBytes();
    public static final byte[] COLUMN_TRAN_ACT_NO_BYTES = "TRAN_ACT_NO".getBytes();
    public static final byte[] COLUMN_PROD_CH_CODE_BYTES = "PROD_CH_CODE".getBytes();
    public static final byte[] COLUMN_PROD_CH_DESC_BYTES = "PROD_CH_DESC".getBytes();
    public static final byte[] COLUMN_SRC_SYS_BYTES = "SRC_SYS".getBytes();

    public static final String OLD_OLD_ACCOUNT_TYPE = "3";
    public static final String OLD_OLD_ACCOUNT_TYPE_TOO = "4";
    public static final String CONTINUITY_FLAG = "0";
    public static final String NOT_CONTINUITY_FLAG = "1";

    final SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");

    @Autowired
    private HbaseTemplate hbaseTemplate;

    public String handle(final String fullTextRequestStr, HttpServletRequest request) {
        logger.info(String.format("fullTextRequestStr: %s", fullTextRequestStr));

        //update by 周秋福 start //增加原因：使得接口适应只返回授权字段的需求
        FullTextRequest fullTextRequest = FullTextUtils.getFullTextRequest(FullTextRequest.class, fullTextRequestStr);
        String clientIp = request.getRemoteAddr();
        String key = clientIp + Constants.HISTORY_TRANS_QUERY_HANDLER_MODULE.toUpperCase();
        //查询授权字段
        Map<String, List<String>> fieldPermissionsCacheMap = FieldPermissionsCache.getFieldPermissionsCacheMap();//全大写key is ip+interfaceName
        List<String> authorizationFieldList = fieldPermissionsCacheMap.get(key);
        //update by 周秋福 end

        List<LscxTranInfo> lscxTranInfoList = null;

        if(authorizationFieldList != null && !authorizationFieldList.isEmpty()){
            final String accNoOrCardNo = fullTextRequest.getAccNoOrCardNo();
            final Date transDateBeg = fullTextRequest.getTransDateBeg();
            final Date transDateEnd = fullTextRequest.getTransDateEnd();
            final String queryField = fullTextRequest.getQueryField();

            logger.info(String.format("accNoOrCardNo: %s, dateBeg: %s, dateEnd: %s"
                    , accNoOrCardNo, transDateBeg!=null?sdf.format(transDateBeg):"", transDateEnd!=null?sdf.format(transDateEnd):""));

            if(StringUtils.isBlank(accNoOrCardNo)) {
                logger.info("accNoOrCardNo is empty!");
                throw new RequestXmlException("accNoOrCardNo is empty!");
            }
            final List<AccountInfo> accountInfoList = new ArrayList<>();

            //将号码当作卡号查询
            logger.info("ByCardNo:Beg");
            List<AccountInfo> accountInfoByCardNoList = queryAccountInfoByCardNo(accNoOrCardNo);
            accountInfoList.addAll(accountInfoByCardNoList);
            accountInfoList.addAll(queryOldActNoAccountInfo(accountInfoByCardNoList));
            accountInfoList.addAll(queryOldOldActNoAccountInfo(accountInfoByCardNoList));
            logger.info(String.format("ByCardNo:End, accountInfoList size: %d", accountInfoList.size()));

            //将号码当作ATM主账号查询
            logger.info("ByAtmAccount:Beg");
            List<AccountInfo> accountInfoByAtmAccountList = queryAccountInfoByAtmAccount(accNoOrCardNo);
            accountInfoList.addAll(accountInfoByAtmAccountList);
            accountInfoList.addAll(queryOldActNoAccountInfo(accountInfoByAtmAccountList));
            accountInfoList.addAll(queryOldOldActNoAccountInfo(accountInfoByAtmAccountList));
            logger.info(String.format("ByAtmAccount:End, accountInfoList size: %d", accountInfoList.size()));

            //将号码当作旧账号查询
            logger.info("ByOldActNo:Beg");
            List<AccountInfo> accountInfoByOldActNoList = queryInfoByOldActNo(accNoOrCardNo);
            accountInfoList.addAll(queryOldOldActNoAccountInfo2(accountInfoByOldActNoList));
            accountInfoList.addAll(accountInfoByOldActNoList);
            logger.info(String.format("ByOldActNo:End, accountInfoList size:%d", accountInfoList.size()));

            //将号码当作旧旧账号查询
            logger.info("ByOldOldActNo:Beg");
            List<AccountInfo> accountInfoByOldOldActNoList = queryInfoByOldOldActNo(accNoOrCardNo);
            accountInfoList.addAll(accountInfoByOldOldActNoList);
            accountInfoList.addAll(queryNewActNoAccountInfo(accountInfoByOldOldActNoList));
            logger.info(String.format("ByOldOldActNo:End, accountInfoList size: %d", accountInfoList.size()));

            lscxTranInfoList = queryLscxTran(accountInfoList, transDateBeg, transDateEnd, authorizationFieldList, queryField);
            logger.info(String.format("lscxTranInfoList size: %d", lscxTranInfoList.size()));
        }

        final FullTextResponse fullTextResponse = new FullTextResponse();
        fullTextResponse.setLscxTranInfoList(lscxTranInfoList);

        return FullTextUtils.getFullTextResponseStr(fullTextResponse);
    }

    /**
     * 查询是否新线卡号
     */
    private List<AccountInfo> queryAccountInfoByCardNo(String cardNo) {
		logger.info(String.format("queryAccountInfoByCardNo Beg:LSCX_BANCS_CARD_ACT_KEY_CARD,cardNo: %s", cardNo));
        Scan scan = new Scan();
        scan.setRowPrefixFilter(cardNo.getBytes());
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_ATM_ACCOUNT_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SUB_ACT_NO_BYTES);
		
		List<AccountInfo> resultList = hbaseTemplate.find(TABLE_LSCX_BANCS_CARD_ACT_KEY_CARD, scan, new RowMapper<AccountInfo>() {
            public AccountInfo mapRow(Result result, int i) throws Exception {
                AccountInfo accountInfo = new AccountInfo();
                accountInfo.setNewActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_ATM_ACCOUNT_BYTES)));
                accountInfo.setNewSubActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SUB_ACT_NO_BYTES)));
                return accountInfo;
            }
        });
		logger.info(String.format("queryAccountInfoByCardNo:End, size: %d", resultList.size()));
        return resultList;
    }

    /**
     * 查询是否新线主账户
     */
    private List<AccountInfo> queryAccountInfoByAtmAccount(String atmAccountNo) {
		logger.info(String.format("queryAccountInfoByAtmAccount Beg: LSCX_MAIN_SUB_ACT_KEY_ACT, atmAccountNo: %s", atmAccountNo));
        Scan scan = new Scan();
		scan.setRowPrefixFilter(atmAccountNo.getBytes());
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_AC_NO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SUB_AC_NO_BYTES);
		List<AccountInfo> resultList = hbaseTemplate.find(TABLE_LSCX_MAIN_SUB_ACT_KEY_ACT, scan, new RowMapper<AccountInfo>() {
			public AccountInfo mapRow(Result result, int i) throws Exception {
				AccountInfo accountInfo = new AccountInfo();
				accountInfo.setNewActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_AC_NO_BYTES)));
				accountInfo.setNewSubActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SUB_AC_NO_BYTES)));
				return accountInfo;
            }
        });
		
		logger.info(String.format("queryAccountInfoByAtmAccount End,size: %d", resultList.size()));
        return resultList;
    }

    /**
     * 查询是否旧线账号
     */
    private List<AccountInfo> queryInfoByOldActNo(final String oldActNo) {
		logger.info(String.format("queryInfoByOldActNo Beg: LSCX_OLD_NEW_ACT_KEY_OLD_ACT, oldActNo: %s", oldActNo));
        Scan scan = new Scan();
		scan.setRowPrefixFilter(oldActNo.getBytes());
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_ACT_NO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_SUB_ACT_NO_BYTES);
		
		List<AccountInfo> resultList = hbaseTemplate.find(TABLE_LSCX_OLD_NEW_ACT_KEY_OLD_ACT, scan, new RowMapper<AccountInfo>() {
                    public AccountInfo mapRow(Result result, int i) throws Exception {
            AccountInfo accountInfo = new AccountInfo();
            final String newActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_ACT_NO_BYTES));
            accountInfo.setNewActNo(newActNo);
            accountInfo.setNewSubActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_SUB_ACT_NO_BYTES)));

            String cOldActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES));
            if (StringUtils.isNotBlank(cOldActNo) && cOldActNo.length() == 18 && oldActNo.matches(".*0188.*")) {
                accountInfo.setOldActNo(cOldActNo.substring(0, 7) + "880101" + cOldActNo.substring(11, 18) + "00");
            }  else {
				accountInfo.setOldActNo(oldActNo);
			}
			return accountInfo;
            }
        });
		logger.info(String.format("queryInfoByOldActNo End, size: %d", resultList.size()));
        return resultList;
    }

    /**
     * 根据旧旧线账号查询
     *
     * @param oldOldActNo
     * @return
     */
    private List<AccountInfo> queryInfoByOldOldActNo(String oldOldActNo) {
		logger.info(String.format("queryInfoByOldOldActNo Beg: LSCX_SAFE_TO_RBS_KEY_OLD_OLD_ACT, oldOldActNo: %s", oldOldActNo));
        Scan scan = new Scan();
		scan.setRowPrefixFilter(oldOldActNo.getBytes());
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_RBS_ACT_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SAFE_ACT_BYTES);
		List<AccountInfo> resultList = hbaseTemplate.find(TABLE_LSCX_SAFE_TO_RBS_KEY_OLD_OLD_ACT, scan, new RowMapper<AccountInfo>() {
			public AccountInfo mapRow(Result result, int i) throws Exception {
				final String oldActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_RBS_ACT_BYTES));
				final String cOldOldActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SAFE_ACT_BYTES));
				AccountInfo accountInfo = new AccountInfo();
				accountInfo.setOldOldActNo(cOldOldActNo);
				if (StringUtils.isNotBlank(oldActNo) && oldActNo.length() == 18 && oldActNo.matches(".*0188.*")) {
					accountInfo.setOldActNo(oldActNo.substring(0, 7) + "880101" + oldActNo.substring(11, 18) + "00");
				} else {
					accountInfo.setOldActNo(oldActNo);
				}
				return accountInfo;
            }
        });
		logger.info(String.format("queryInfoByOldOldActNo End, size: %d", resultList.size()));
        return resultList;
    }

    private List<AccountInfo> queryNewActNoAccountInfo(List<AccountInfo> accountInfoList) {
		logger.info(String.format("queryNewActNoAccountInfo Beg: LSCX_OLD_NEW_ACT_KEY_OLD_ACT, accountInfoList size: %d", accountInfoList.size()));

        List<AccountInfo> newActNoAccountInfoList = new ArrayList<>();
        for (final AccountInfo accountInfo : accountInfoList) {
            final String oldActNo = accountInfo.getOldActNo();
            logger.info(String.format("accountInfo oldActNo: %s", oldActNo));
            Scan scan = new Scan();
			scan.setRowPrefixFilter(oldActNo.getBytes());
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_ACT_NO_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_SUB_ACT_NO_BYTES);

            final List<AccountInfo> tmpList = hbaseTemplate.find(TABLE_LSCX_OLD_NEW_ACT_KEY_OLD_ACT, scan, new RowMapper<AccountInfo>() {
                public AccountInfo mapRow(Result result, int i) throws Exception {
                final String actType = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES));
                final String newActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_ACT_NO_BYTES));
                final String newSubActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_SUB_ACT_NO_BYTES));

                AccountInfo tmpAccountInfo = null;
                if (actType.equals(OLD_OLD_ACCOUNT_TYPE) || actType.equals(OLD_OLD_ACCOUNT_TYPE_TOO)) {
                    tmpAccountInfo = new AccountInfo();
                    tmpAccountInfo.setNewActNo(newActNo);
                    tmpAccountInfo.setNewSubActNo(newSubActNo);
                }
                return tmpAccountInfo;
                }
            });
			logger.info(String.format("accountInfo newActNo: %s found, before trimNullVal size: %d", accountInfo.getNewActNo(), tmpList.size()));
            CommonUtils.trimCollectionNullVal(tmpList);
            logger.info(String.format("accountInfo newActNo: %s found, after trimNullVal size: %d", accountInfo.getNewActNo(), tmpList.size()));
            newActNoAccountInfoList.addAll(tmpList);
        }
		logger.info(String.format("queryNewActNoAccountInfo End,size:", newActNoAccountInfoList.size()));
        return newActNoAccountInfoList;
    }

    private List<AccountInfo> queryOldActNoAccountInfo(List<AccountInfo> accountInfoList) {
        logger.info(String.format("queryOldActNoAccountInfo Beg: LSCX_OLD_NEW_ACT_KEY_NEW_ACT, accountInfoList size: %d", accountInfoList.size()));

        List<AccountInfo> fillOldActNoAccountInfoList = new ArrayList<>();
        for (final AccountInfo accountInfo : accountInfoList) {
            logger.info(String.format("accountInfo newActNo + newSubActNo: %s", (accountInfo.getNewActNo() + accountInfo.getNewSubActNo())));

            Scan scan = new Scan();
            //TO-DO hbase建表语句确实用主账号子账号联合当key，但doc文档并没有这么说 20170620文档已增加说明
			scan.setRowPrefixFilter((accountInfo.getNewActNo() + accountInfo.getNewSubActNo()).getBytes());
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_ACT_NO_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_SUB_ACT_NO_BYTES);

            final List<AccountInfo> tmpList = hbaseTemplate.find(TABLE_LSCX_OLD_NEW_ACT_KEY_NEW_ACT, scan, new RowMapper<AccountInfo>() {
                public AccountInfo mapRow(Result result, int i) throws Exception {
                    final String actType = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES));
                    final String newActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_NEW_ACT_NO_BYTES));
                    final String oldActNo = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES));

                    AccountInfo tmpAccountInfo = null;
                    if (!OLD_OLD_ACCOUNT_TYPE.equals(actType)) {
						tmpAccountInfo = new AccountInfo();
                        if (StringUtils.isNotBlank(oldActNo) && oldActNo.length() == 18 && oldActNo.matches(".*0188.*")) {
                            tmpAccountInfo.setOldActNo(oldActNo.substring(0, 7) + "880101" + oldActNo.substring(11, 18) + "00");
                        } else if (StringUtils.isNotBlank(newActNo) && newActNo.length() > 17 && newActNo.matches(".*0188.*")) {
                            tmpAccountInfo.setOldActNo(newActNo.substring(0, 7) + "880101" + newActNo.substring(11, 18) + "00");
                        } else if (StringUtils.isNotBlank(newActNo) && newActNo.length() > 17) {
                            tmpAccountInfo.setOldActNo(newActNo);
                        } else {
                            tmpAccountInfo.setOldActNo(oldActNo);
                        }
                    }
					//TODO else???
                    return tmpAccountInfo;
                }
            });

			logger.info(String.format("accountInfo newActNo: %s before trim null val found size: %d", accountInfo.getNewActNo(), tmpList.size()));
			
            CommonUtils.trimCollectionNullVal(tmpList);
			
            logger.info(String.format("accountInfo newActNo: %s after trim null val found size: %d", accountInfo.getNewActNo(), tmpList.size()));
            fillOldActNoAccountInfoList.addAll(tmpList);
        }
		logger.info(String.format("queryOldActNoAccountInfo End,size: %d", fillOldActNoAccountInfoList.size()));
        return fillOldActNoAccountInfoList;
    }

    private List<AccountInfo> queryOldOldActNoAccountInfo(List<AccountInfo> accountInfoList) {
        logger.info(String.format("queryOldOldActNoAccountInfo Beg: LSCX_OLD_NEW_ACT_KEY_NEW_ACT, accountInfoList size: %d", accountInfoList.size()));

        List<AccountInfo> oldOldActNoAccountInfoList = new ArrayList<>();
        for (final AccountInfo accountInfo : accountInfoList) {
            //TO-DO hbase建表语句确实用主账号子账号联合当key，但doc文档并没有这么说  20160620文档已增加说明
            logger.info(String.format("accountInfo newActNo+newSubActNo: %s", (accountInfo.getNewActNo()+accountInfo.getNewSubActNo())));

            Scan scan = new Scan();
			scan.setRowPrefixFilter((accountInfo.getNewActNo()+accountInfo.getNewSubActNo()).getBytes());
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES);

            final List<AccountInfo> tmpList = hbaseTemplate.find(TABLE_LSCX_OLD_NEW_ACT_KEY_NEW_ACT, scan, new RowMapper<AccountInfo>() {
                public AccountInfo mapRow(Result result, int i) throws Exception {
                    final String actType = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_TYPE_BYTES));
                    AccountInfo tmpAccountInfo = null;
                    if (actType.equals(OLD_OLD_ACCOUNT_TYPE)) {
						tmpAccountInfo = new AccountInfo();
                        tmpAccountInfo.setOldOldActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_OLD_ACT_NO_BYTES)));
                    }//TODO else???
                    return tmpAccountInfo;
                }
            });
            CommonUtils.trimCollectionNullVal(tmpList);
            logger.info(String.format("accountInfo newActNo+newSubActNo: %s found size: %d", (accountInfo.getNewActNo()+accountInfo.getNewSubActNo()), tmpList.size()));
            oldOldActNoAccountInfoList.addAll(tmpList);
        }
		logger.info(String.format("queryOldOldActNoAccountInfo End,size: %d", oldOldActNoAccountInfoList.size()));
        return oldOldActNoAccountInfoList;
    }

    private List<AccountInfo> queryOldOldActNoAccountInfo2(List<AccountInfo> accountInfoList) {
        logger.info(String.format("queryOldOldActNoAccountInfo2 Beg: LSCX_SAFE_TO_RBS_KEY_OLD_ACT, accountInfoList size: %d", accountInfoList.size()));

        List<AccountInfo> oldOldActNoAccountInfoList = new ArrayList<>();
        for (final AccountInfo accountInfo : accountInfoList) {
            logger.info(String.format("accountInfo oldActNo: %s", accountInfo.getOldActNo()));

            Scan scan = new Scan();
			scan.setRowPrefixFilter(accountInfo.getOldActNo().getBytes());
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_RBS_ACT_BYTES);
            scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SAFE_ACT_BYTES);
            final List<AccountInfo> tmpList = hbaseTemplate.find(TABLE_LSCX_SAFE_TO_RBS_KEY_OLD_ACT, scan, new RowMapper<AccountInfo>() {
                public AccountInfo mapRow(Result result, int i) throws Exception {
                    AccountInfo tmpAccountInfo = new AccountInfo();
                    tmpAccountInfo.setOldOldActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SAFE_ACT_BYTES)));
                    return tmpAccountInfo;
                }
            });
            logger.info(String.format("accountInfo oldActNo: %s found size: %d", accountInfo.getOldActNo(), tmpList.size()));
            oldOldActNoAccountInfoList.addAll(tmpList);
        }
		
		logger.info(String.format("queryOldOldActNoAccountInfo2 End, size: %d", oldOldActNoAccountInfoList.size()));
        return oldOldActNoAccountInfoList;
    }

    private List<LscxTranInfo> queryLscxTran(List<AccountInfo> accountInfoList, final Date dateBeg, final Date dateEnd, List<String> authorizationFieldList, final String queryField) {
        //分成四组，分别去重
        Set<String> newActNoSet = new HashSet<>();
        Set<String> newSubActNoSet = new HashSet<>();
        Set<String> oldActNoSet = new HashSet<>();
        Set<String> oldOldActNoSet = new HashSet<>();
        
        for(AccountInfo accountInfo : accountInfoList) {
            final String newActNo = accountInfo.getNewActNo();
            final String newSubActNo = accountInfo.getNewSubActNo();
            final String oldActNo = accountInfo.getOldActNo();
            final String oldOldActNo = accountInfo.getOldOldActNo();

            if(StringUtils.isNotBlank(newActNo)) newActNoSet.add(newActNo);
            if(StringUtils.isNotBlank(newSubActNo)) newSubActNoSet.add(newSubActNo);
            if(StringUtils.isNotBlank(oldActNo)) oldActNoSet.add(oldActNo);
            if(StringUtils.isNotBlank(oldOldActNo)) oldOldActNoSet.add(oldOldActNo);
        }

        List<LscxTranInfo> lscxTranInfoList = new ArrayList<>();
		logger.info("newActNoSet:" + newActNoSet);
		logger.info("newSubActNoSet:" + newSubActNoSet);
		logger.info("oldActNoSet:" + oldActNoSet);
		logger.info("oldOldActNoSet:" + oldOldActNoSet);
        for (String accNo : newActNoSet) {
            queryLscxInfoAndCheckContinuity(accNo, dateBeg, dateEnd, lscxTranInfoList, true, authorizationFieldList, queryField);
        }
        for (String accNo : newSubActNoSet) {
            queryLscxInfoAndCheckContinuity(accNo, dateBeg, dateEnd, lscxTranInfoList, true, authorizationFieldList, queryField);
        }
        for (String accNo : oldActNoSet) {
            queryLscxInfoAndCheckContinuity(accNo, dateBeg, dateEnd, lscxTranInfoList, false, authorizationFieldList, queryField);
        }
        for (String accNo : oldOldActNoSet) {
            queryLscxInfoAndCheckContinuity(accNo, dateBeg, dateEnd, lscxTranInfoList, false, authorizationFieldList, queryField);
        }

        return lscxTranInfoList;
    }

    private void queryLscxInfoAndCheckContinuity(String actNo, final Date dateBeg, final Date dateEnd, List<LscxTranInfo> lscxTranInfoList,
                                                 boolean checkContinuity, final List<String> authorizationFieldList, final String queryField) {
        logger.info(String.format("queryLscxInfoAndCheckContinuity Beg: LSCX_TRAN, actNo: %s, dateBeg: %s, dateEnd: %s"
                , actNo, dateBeg!=null?sdf.format(dateBeg):"", dateEnd!=null?sdf.format(dateEnd):""));

        Scan scan = new Scan();
		scan.setRowPrefixFilter(actNo.getBytes());
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_NO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_REC_NO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_APPG_DATE_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_POST_DATE_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SYS_DATE_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SYS_TIME_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_BR_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_TELLER_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_CURR_CODE_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_CURR_NAME_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_NAME_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_AMT_BYTES);
		scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_BAL_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_PROD_CQ_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_MEMO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_ACT_NO_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_PROD_CH_CODE_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_PROD_CH_DESC_BYTES);
        scan.addColumn(COLUMN_FAMILY_F_BYTES, COLUMN_SRC_SYS_BYTES);

        final List<LscxTranInfo> tmpList = hbaseTemplate.find(TABLE_LSCX_TRAN, scan, new RowMapper<LscxTranInfo>() {
            public LscxTranInfo mapRow(Result result, int i) throws Exception {


                String dateStr = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_APPG_DATE_BYTES));

				if(StringUtils.isBlank(dateStr)) {
					dateStr = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SYS_DATE_BYTES));
				}
				if(StringUtils.isBlank(dateStr)) {
					dateStr = CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_POST_DATE_BYTES));
				}
				Date appgDate = sdf.parse(dateStr);
				if(dateBeg!=null && appgDate.before(dateBeg)) {
					return null;
				}
				if(dateEnd!=null && appgDate.after(dateEnd)) {
					return null;
				}

				//update by 周秋福 start //修改原因：对字段增加权限控制
                LscxTranInfo lscxTranInfo = new LscxTranInfo();

                final int ONE_RESULT = 1;
                final String ALL_FIELD = "*";
				if(authorizationFieldList.size() == ONE_RESULT){
				    if(authorizationFieldList.contains(ALL_FIELD)){
				        List<String> list = new ArrayList<String>();

                        //list.add(Constants.HISTORY_TRANS_QUERY_MODULE_ACT_NO);
                        //list.add(Constants.HISTORY_TRANS_QUERY_MODULE_REC_NO);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_AMT);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_BAL);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_APPG_DATE);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_SYS_DATE);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_SYS_TIME);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_POST_DATE);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_SRC_SYS);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_BR);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_TELLER);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_CURR_CODE);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_CURR_NAME);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_NAME);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_PROD_CQ);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_ACT_NO);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_PROD_CH_CODE);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_PROD_CH_DESC);
                        //list.add(Constants.HISTORY_TRANS_QUERY_MODULE_CONTINUITY);
                        list.add(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_MEMO);

                        authorizationFieldList.clear();//清空所有元素
                        authorizationFieldList.addAll(list);
                    }
                }

                if(StringUtils.isNotBlank(queryField)){//表示需要进一步缩小查询字段的范围
                    List<String> queryFieldList = new ArrayList<String>();
                    if(queryField.contains(separator)){//包含分隔符
                        String[] queryFieldArray = queryField.split(separator);
                        for(String element : queryFieldArray){
                            if(StringUtils.isNotBlank(element)){
                                queryFieldList.add(element.trim().toUpperCase());
                            }
                        }
                    }else{//某个字段的名称
                        queryFieldList.add(queryField.trim().toUpperCase());
                    }
                    authorizationFieldList.clear();//清空所有元素
                    authorizationFieldList.addAll(queryFieldList);
                }

                lscxTranInfo.setActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_NO_BYTES)));

                /*
				if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_ACT_NO)){
                    lscxTranInfo.setActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_ACT_NO_BYTES)));
                }
                */

                lscxTranInfo.setRecNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_REC_NO_BYTES)));
                /*
                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_REC_NO)){
                    lscxTranInfo.setRecNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_REC_NO_BYTES)));
                }
                */

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_AMT)){
                    lscxTranInfo.setTranAmt(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_AMT_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_BAL)){
                    lscxTranInfo.setTranBal(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_BAL_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_APPG_DATE)){
                    lscxTranInfo.setAppgDate(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_APPG_DATE_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_SYS_DATE)){
                    lscxTranInfo.setSysDate(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SYS_DATE_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_SYS_TIME)){
                    lscxTranInfo.setSysTime(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SYS_TIME_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_POST_DATE)){
                    lscxTranInfo.setPostDate(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_POST_DATE_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_SRC_SYS)){
                    lscxTranInfo.setSrcSys(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_SRC_SYS_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_BR)){
                    lscxTranInfo.setTranBr(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_BR_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_TELLER)){
                    lscxTranInfo.setTranTeller(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_TELLER_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_CURR_CODE)){
                    lscxTranInfo.setCurrCode(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_CURR_CODE_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_CURR_NAME)){
                    lscxTranInfo.setCurrName(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_CURR_NAME_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_NAME)){
                    lscxTranInfo.setTranMemo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_NAME_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_PROD_CQ)){
                    lscxTranInfo.setProdCq(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_PROD_CQ_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_TRAN_ACT_NO)){
                    lscxTranInfo.setTranActNo(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_TRAN_ACT_NO_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_PROD_CH_CODE)){
                    lscxTranInfo.setProdChCode(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_PROD_CH_CODE_BYTES)));
                }

                if(authorizationFieldList.contains(Constants.HISTORY_TRANS_QUERY_MODULE_PROD_CH_DESC)){
                    lscxTranInfo.setProdChDesc(CommonUtils.byteArray2String(result.getValue(COLUMN_FAMILY_F_BYTES, COLUMN_PROD_CH_DESC_BYTES)));
                }
                //update by 周秋福 end
                return lscxTranInfo;
            }
        });

		List<LscxTranInfo> tmpNotNullList = new ArrayList<>();
		for(LscxTranInfo tmp : tmpList) {
			if(tmp != null) tmpNotNullList.add(tmp);
		}

		if(tmpNotNullList != null && !tmpNotNullList.isEmpty()){
            if(checkContinuity) checkLscxContinuity(tmpNotNullList);
            lscxTranInfoList.addAll(tmpNotNullList);
        }

		logger.info(String.format("queryLscxInfoAndCheckContinuity End, size: %d", tmpNotNullList.size()));

    }

    //只检测主子账号的时候
    private void checkLscxContinuity(List<LscxTranInfo> lscxTranInfoList) {
        //按照记录号（rec_no）降序和数据日期(appg_date)升序进行排序
        Collections.sort(lscxTranInfoList, new Comparator<LscxTranInfo>() {
            public int compare(LscxTranInfo o1, LscxTranInfo o2) {
                int flag = o1.getActNo().compareTo(o2.getActNo());
                if(flag != 0) return flag;
                else return o2.getRecNo().compareTo(o1.getRecNo());
            }
        });

        LscxTranInfo preLscxTranInfo = null; //用于检验是否已经是另一个actNo了
        for(int i = 0; i< lscxTranInfoList.size(); i++) {
            LscxTranInfo lscxTranInfo = lscxTranInfoList.get(i);

            if(preLscxTranInfo==null || preLscxTranInfo!=null && !preLscxTranInfo.getActNo().equals(lscxTranInfo.getActNo())) {
                //连续性针对同一个账号的交易信息进行校验  !preLscxTranInfo.getActNo().equals(lscxTranInfo.getActNo())
                lscxTranInfo.setContinuity(CONTINUITY_FLAG);
            } else {
                preLscxTranInfo = lscxTranInfoList.get(i - 1);
				String tranAmtStr = lscxTranInfo.getTranAmt();
				String tranBalStr = lscxTranInfo.getTranBal();
				String preLscxTranInfoTranBalStr = preLscxTranInfo.getTranBal();
				if((StringUtils.isBlank(tranAmtStr) 
					|| StringUtils.isBlank(tranBalStr) 
					|| StringUtils.isBlank(preLscxTranInfoTranBalStr))
					|| !(new BigDecimal(tranBalStr).subtract(new BigDecimal(tranAmtStr))
							.equals(new BigDecimal(preLscxTranInfoTranBalStr)))) {
					lscxTranInfo.setContinuity(NOT_CONTINUITY_FLAG);
					preLscxTranInfo.setContinuity(NOT_CONTINUITY_FLAG);
				} else {
					lscxTranInfo.setContinuity(CONTINUITY_FLAG);
				}
                
            }
            preLscxTranInfo = lscxTranInfo;
        }
    }
}
